package com.atguigu.mq.listener;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * @author zhumq
 */ // 定义一个消息监听器类，用于处理从消息队列中接收到的消息
@Component
@Slf4j
public class MyMessageListener {
    
    // 定义常量，表示不同类型的队列名称
    public static final String QUEUE_NAME  = "queue.order";
    public static final String QUEUE_NORMAL = "queue.normal.video";
    public static final String QUEUE_DEAD_LETTER = "queue.dead.letter.video";
    public static final String QUEUE_DELAY = "queue.test.delay";
    
    /**
     * 处理从普通队列接收到的消息
     *
     * @param dataString 消息的内容，以字符串形式表示
     * @param message 消息对象，包含消息的详细信息
     * @param channel 信道对象，用于与消息队列进行交互
     * @throws IOException 如果处理过程中发生I/O异常
     */
    // @RabbitListener(queues = {QUEUE_NAME})
    public void processMessage(String dataString, Message message, Channel channel) throws IOException {

        // 获取当前消息的 deliveryTag , 用于确认消息处理状态
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            // 核心操作：记录消息内容
            log.info("消费端 消息内容：{}" , dataString);
            
            // 示例操作，用于模拟核心操作中的异常情况
            System.out.println(10 / 0);

            // 核心操作成功：返回 ACK 信息 ，向消息队列发送确认信息，表示消息已成功处理
            channel.basicAck(deliveryTag, false);

        } catch (Exception e) {

            // 获取当前消息是否是重复投递的
            //      redelivered 为 true：说明当前消息已经重复投递过一次了
            //      redelivered 为 false：说明当前消息是第一次投递
            Boolean redelivered = message.getMessageProperties().getRedelivered();

            // 核心操作失败：返回 NACK 信息
            // requeue 参数：控制消息是否重新放回队列
            //      取值为 true：重新放回队列，broker 会重新投递这个消息
            //      取值为 false：不重新放回队列，broker 会丢弃这个消息
            
            // 根据消息是否重复投递，决定是否将消息重新放回队列
            if (redelivered) {
                // 只重新投递一次
                // 如果当前消息已经是重复投递的，说明此前已经重试过一次，所以 requeue 设置为 false，表示不重新放回队列
                channel.basicNack(deliveryTag, false, false);
            } else {
                // 如果当前消息是第一次投递，说明当前代码是第一次抛异常，尚未重试，所以 requeue 设置为 true，表示重新放回队列在投递一次
                channel.basicNack(deliveryTag, false, true);
            }

            // reject 表示拒绝
            // 辨析：basicNack() 和 basicReject() 方法区别
            // basicNack()能控制是否批量操作
            // basicReject()不能控制是否批量操作
            // channel.basicReject(deliveryTag, true);
        }
    }
    
    /**
     * 处理测试预取策略的队列中的消息  -- 针对大量数据进行操作
     *
     * @param dataString 消息的内容，以字符串形式表示
     * @param message 消息对象，包含消息的详细信息
     * @param channel 信道对象，用于与消息队列进行交互
     * @throws IOException 如果处理过程中发生I/O异常
     * @throws InterruptedException 如果线程在等待过程中被中断
     */
    @RabbitListener(queues = {QUEUE_NAME})
    public void processMessageTestPrefetch(String dataString, Message message, Channel channel) throws IOException, InterruptedException {
        
        log.info("消费端 消息内容：{}" , dataString);
        
        // 模拟消息处理过程中的延迟
        TimeUnit.SECONDS.sleep(1);
        
        // 确认消息处理成功
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
    
    /**
     * 处理正常队列中的消息，但选择拒绝消息
     *
     * @param message 消息对象，包含消息的详细信息
     * @param channel 信道对象，用于与消息队列进行交互
     * @throws IOException 如果处理过程中发生I/O异常
     */
    @RabbitListener(queues = {QUEUE_NORMAL})
    public void processMessageNormal(Message message, Channel channel) throws IOException {
        
        // 监听正常队列，但是拒绝消息
        log.info("★[normal]消息接收到，但我拒绝。");
        
        // 拒绝消息，不重新放回队列
        channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
    }
    
    
    /**
     * 处理死信队列中的消息
     *
     * @param dataString 消息的内容，以字符串形式表示
     * @param message 消息对象，包含消息的详细信息
     * @param channel 信道对象，用于与消息队列进行交互
     * @throws IOException 如果处理过程中发生I/O异常
     */
    @RabbitListener(queues = {QUEUE_DEAD_LETTER})
    public void processMessageDead(String dataString, Message message, Channel channel) throws IOException {
        // 监听死信队列
        log.info("★[dead letter]dataString = {}", dataString);
        log.info("★[dead letter]我是死信监听方法，我接收到了死信消息");
        
        // 确认消息处理成功
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
    
    
    /**
     * 处理延迟队列中的消息
     *
     * @param dataString 消息的内容，以字符串形式表示
     * @param message 消息对象，包含消息的详细信息
     * @param channel 信道对象，用于与消息队列进行交互
     * @throws IOException 如果处理过程中发生I/O异常
     */
    @RabbitListener(queues = {QUEUE_DELAY})
    public void processMessageDelay(String dataString, Message message, Channel channel) throws IOException {
        log.info("[delay message][消息本身]{}", dataString);
        log.info("[delay message][当前时间]{}", new SimpleDateFormat("HH:mm:ss").format(new Date()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
    
    
    // 定义常量，表示优先级队列的名称
    public static final String QUEUE_PRIORITY = "queue.test.priority";
    
    /**
     * 处理优先级队列中的消息
     *
     * @param dataString 消息的内容，以字符串形式表示
     * @param message 消息对象，包含消息的详细信息
     * @param channel 信道对象，用于与消息队列进行交互
     * @throws IOException 如果处理过程中发生I/O异常
     */
    @RabbitListener(queues = {QUEUE_PRIORITY})
    public void processMessagePriority(String dataString, Message message, Channel channel) throws IOException {
        log.info("[priority]{}", dataString);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}